package com.zx.mes.coolmq.hyl.config;

import com.rabbitmq.client.Channel;
import com.zx.mes.coolmq.hyl.util.MqConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

/**
 * RabbitMQ交换机、队列的配置类.定义交换机、key、queue并做好绑定。
 * 同时定义每个队列的ttl，队列最大长度，Qos等等
 * 这里只绑定了死信队列。建议每个队列定义自己的QueueConfig
 * @author hyl
 * @date 2018-3-19s
 * @version 1.0.0
 */
@Configuration
public class DeadQueueConfig {

    @Component
    public class DeadLetterMessageListener implements ChannelAwareMessageListener{
        private  final Logger logger = LoggerFactory.getLogger(DeadLetterMessageListener.class);

        @Autowired
        private RedisTemplate redisTemplate;

        @Autowired
        private StringRedisTemplate stringRedisTemplate;

        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            MessageProperties messageProperties = message.getMessageProperties();
            //消息体
            String messageBody = new String(message.getBody());

            logger.warn("dead letter message：{} | tag：{}",messageBody,message.getMessageProperties().getDeliveryTag());

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            stringRedisTemplate.delete(messageProperties.getMessageId());

//            redisTemplate.opsForHash().delete(MqConstants.MQ_CONSUMER_RETRY_COUNT_KEY, messageProperties.getMessageId());
        }
    }

    //========================== 声明交换机 ==========================

    /**
     * 死信交换机
     */
    @Bean
    public DirectExchange dlxExchange(){
        return new DirectExchange(MqConstants.DLX_EXCHANGE);
    }

    //========================== 声明队列 ===========================

    @Bean
    public Queue dlxQueue(){
        return new Queue(MqConstants.DLX_QUEUE, true, false, false);
    }

    /**
     * 通过死信路由key绑定死信交换机和死信队列
     */
    @Bean
    public Binding dlxBinding(){
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
                .with(MqConstants.DLX_ROUTING_KEY);
    }


    /**
     * 死信队列的监听
     * @param connectionFactory RabbitMQ连接工厂
     * @param deadLetterMessageListener 死信的监听者
     * @return 监听容器对象
     */
    @Bean
    public SimpleMessageListenerContainer deadLetterListenerContainer(ConnectionFactory connectionFactory,
                                                                      DeadLetterMessageListener deadLetterMessageListener) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(dlxQueue());
        container.setExposeListenerChannel(true);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(deadLetterMessageListener);
        /** 设置消费者能处理消息的最大个数 */
        container.setPrefetchCount(100);
        return container;
    }
}
